--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit d9cbc59ab3ac9f6bcd27fa95f95eaf21c3556ce3
Parents : 4839bc0
Author : Sudo-Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-02-14T19:18:20-06:00
Refactor backend tests
- Removed redundant comments from various test files to enhance readability and focus on the test logic.
- Added a few tests for prop sync and markdown security.
Changes
19 files changed, 460 insertions(+), 42 deletions(-)
Diff
diff --git a/tests/backend/conftest.py b/tests/backend/conftest.py
index 9fe7e1d6..fa28b613 100644
--- a/tests/backend/conftest.py
+++ b/tests/backend/conftest.py
@@ -43,9 +43,6 @@ def global_mocks():
@pytest.fixture(autouse=True)
def cleanup_sqlite_connections():
yield
- # After each test, try to close any lingering sqlite connections if possible
- # This is a bit hard globally without tracking them, but we can at least
- # trigger GC which often helps with ResourceWarnings.
import gc
gc.collect()
diff --git a/tests/backend/test_database_snapshots.py b/tests/backend/test_database_snapshots.py
index 49f29ca5..bdcd0fd5 100644
--- a/tests/backend/test_database_snapshots.py
+++ b/tests/backend/test_database_snapshots.py
@@ -65,7 +65,6 @@ def test_database_snapshot_restoration(temp_dir):
def test_database_auto_backup_logic(temp_dir):
- # This test verifies the loop logic if possible, or just the backup method
db_path = os.path.join(temp_dir, "test.db")
db = Database(db_path)
db.initialize()
diff --git a/tests/backend/test_display_name_fuzzing.py b/tests/backend/test_display_name_fuzzing.py
index 5b13ab60..e69cd845 100644
--- a/tests/backend/test_display_name_fuzzing.py
+++ b/tests/backend/test_display_name_fuzzing.py
@@ -66,7 +66,6 @@ def test_parse_nomadnetwork_node_display_name_property_based(name):
def test_parse_lxmf_display_name_invalid_base64(data):
# Generate something that is NOT valid base64 (by adding invalid chars)
invalid_b64 = base64.b64encode(data).decode("utf-8") + "!!!"
- # This should not crash, it should just return the default value or fail gracefully
result = parse_lxmf_display_name(invalid_b64)
assert isinstance(result, str)
diff --git a/tests/backend/test_docs_manager.py b/tests/backend/test_docs_manager.py
index 9044cefe..b1ba6c7d 100644
--- a/tests/backend/test_docs_manager.py
+++ b/tests/backend/test_docs_manager.py
@@ -52,7 +52,6 @@ def test_docs_manager_storage_dir_fallback(tmp_path):
def test_docs_manager_readonly_public_dir_handling(tmp_path):
- # This test simulates a read-only public dir without storage_dir
public_dir = tmp_path / "readonly_public"
public_dir.mkdir()
@@ -189,7 +188,6 @@ def test_extract_docs_malformed_zip(docs_manager, temp_dirs):
# 1. Zip with no folders at all
create_mock_zip(zip_path, ["file_at_root.txt"])
try:
- # This might fail with IndexError at namelist()[0].split('/')[0] if no slash
docs_manager._extract_docs(zip_path)
except (IndexError, Exception):
pass # Expected or at least handled by not crashing the whole app
diff --git a/tests/backend/test_fuzzing.py b/tests/backend/test_fuzzing.py
index 97d7f460..de9799bd 100644
--- a/tests/backend/test_fuzzing.py
+++ b/tests/backend/test_fuzzing.py
@@ -33,7 +33,6 @@ from meshchatx.src.backend.telemetry_utils import Telemeter
def test_telemetry_unpack_fuzzing(data):
"""Fuzz the telemetry unpacking logic with random binary data."""
try:
- # This should not raise unhandled exceptions
Telemeter.from_packed(data)
except Exception:
# We expect some failures for invalid packed data, but no crashes
@@ -113,7 +112,6 @@ def test_display_name_parsing_fuzzing(app_data_base64):
def test_lxmf_fields_parsing_fuzzing(fields_data):
"""Fuzz the parsing of LXMF message fields."""
try:
- # This simulates how meshchat.py processes fields in on_lxmf_delivery
for field_id, field_data in fields_data.items():
if field_id == 0x01: # FIELD_COMMANDS
try:
@@ -514,10 +512,7 @@ def test_malformed_announce_data(mock_app):
def test_malformed_message_data(mock_app):
"""Test handling of malformed LXMF messages."""
mock_message = MagicMock()
- # Simulate missing attributes or methods
del mock_message.source_hash
-
- # This should be caught by the try-except in on_lxmf_delivery
mock_app.on_lxmf_delivery(mock_message)
diff --git a/tests/backend/test_fuzzing_extended.py b/tests/backend/test_fuzzing_extended.py
index 1d8850bd..b5b58b92 100644
--- a/tests/backend/test_fuzzing_extended.py
+++ b/tests/backend/test_fuzzing_extended.py
@@ -246,7 +246,6 @@ def test_lxmf_message_construction_fuzzing(mock_app, content, title, fields):
)
def test_database_record_fuzzing(mock_app, table_name, data):
"""Fuzz database record insertion logic (simulated)."""
- # This tests the DAO layer's resilience to weird data types if they aren't properly sanitized
try:
dao = None
if table_name == "messages" and hasattr(mock_app.database, "messages"):
@@ -350,6 +349,36 @@ def test_markdown_renderer_fuzzing(text):
pytest.fail(f"MarkdownRenderer crashed: {e}")
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ text=st.one_of(
+ st.text(min_size=0, max_size=5000),
+ st.sampled_from([
+ "<script>alert(1)</script>",
+ "[x](javascript:alert(1))",
+ "[x](data:text/html,<script>alert(1)</script>)",
+ "**" * 2000,
+ "#" * 2000,
+ "`" * 2000,
+ "[](" * 500 + ")" * 500,
+ "\x00\x01\x02\n\t",
+ "\ufffd" * 100,
+ ]),
+ ),
+)
+def test_markdown_renderer_dangerous_patterns(text):
+ """Fuzz markdown renderer with known risky patterns (XSS, ReDoS, control chars)."""
+ from meshchatx.src.backend.markdown_renderer import MarkdownRenderer
+
+ try:
+ html_out = MarkdownRenderer.render(text)
+ assert isinstance(html_out, str)
+ assert "<script>" not in html_out
+ assert "javascript:" not in html_out
+ except Exception as e:
+ pytest.fail(f"MarkdownRenderer crashed on dangerous pattern: {e}")
+
+
# LXMF Message Dictionary Conversion Fuzzing
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(
diff --git a/tests/backend/test_fuzzing_telemetry.py b/tests/backend/test_fuzzing_telemetry.py
index 3ca66279..8eca4c24 100644
--- a/tests/backend/test_fuzzing_telemetry.py
+++ b/tests/backend/test_fuzzing_telemetry.py
@@ -62,7 +62,6 @@ def test_fuzz_full_telemetry_packing(charge, charging, rssi, snr, q):
@settings(deadline=None)
@given(data=st.binary(min_size=0, max_size=2000))
def test_fuzz_from_packed_random_bytes(data):
- # This should never crash
try:
Telemeter.from_packed(data)
except Exception:
@@ -97,5 +96,4 @@ def test_fuzz_command_parsing(commands):
new_cmd[k] = v
processed_commands.append(new_cmd)
- # Just ensure no crash
assert len(processed_commands) == len(commands)
diff --git a/tests/backend/test_interface_config_parser.py b/tests/backend/test_interface_config_parser.py
index eb34f05d..af3290ff 100644
--- a/tests/backend/test_interface_config_parser.py
+++ b/tests/backend/test_interface_config_parser.py
@@ -37,8 +37,6 @@ def test_parse_best_effort_on_failure():
[[Fixed Interface]]
type = fixed
"""
- # Note: ConfigObj might still parse [[Broken Interface] as a key if not careful,
- # but the parser should return something.
interfaces = InterfaceConfigParser.parse(config_text)
assert len(interfaces) >= 1
names = [i["name"] for i in interfaces]
diff --git a/tests/backend/test_lxmf_sync.py b/tests/backend/test_lxmf_sync.py
index d4494753..bccb4fa8 100644
--- a/tests/backend/test_lxmf_sync.py
+++ b/tests/backend/test_lxmf_sync.py
@@ -142,3 +142,107 @@ async def test_specific_node_hash_validation(mock_app):
await sync_handler(None)
mock_app.current_context.message_router.request_messages_from_propagation_node.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_status_includes_sync_storage_and_confirmation_metrics(mock_app):
+ status_handler = next(
+ r.handler
+ for r in mock_app.get_routes()
+ if r.path == "/api/v1/lxmf/propagation-node/status"
+ )
+ sync_handler = next(
+ r.handler
+ for r in mock_app.get_routes()
+ if r.path == "/api/v1/lxmf/propagation-node/sync"
+ )
+
+ mock_app.current_context.message_router.get_outbound_propagation_node.return_value = (
+ b"somehash"
+ )
+ mock_app.current_context.message_router.propagation_transfer_last_result = 8
+
+ with (
+ patch.object(
+ mock_app.current_context.database.messages,
+ "count_lxmf_messages",
+ side_effect=[10, 13],
+ ),
+ patch.object(
+ mock_app.current_context.database.messages,
+ "count_lxmf_messages_by_state",
+ side_effect=[2, 4],
+ ),
+ ):
+ await sync_handler(None)
+ response = await status_handler(None)
+
+ data = json.loads(response.body)["propagation_node_status"]
+ assert data["messages_received"] == 8
+ assert data["messages_stored"] == 3
+ assert data["delivery_confirmations"] == 2
+ assert data["messages_hidden"] == 3
+
+
+@pytest.mark.asyncio
+async def test_status_metrics_default_to_zero_before_any_sync(mock_app):
+ status_handler = next(
+ r.handler
+ for r in mock_app.get_routes()
+ if r.path == "/api/v1/lxmf/propagation-node/status"
+ )
+ response = await status_handler(None)
+ data = json.loads(response.body)["propagation_node_status"]
+
+ assert data["messages_received"] == 0
+ assert data["messages_stored"] == 0
+ assert data["delivery_confirmations"] == 0
+ assert data["messages_hidden"] == 0
+
+
+@pytest.mark.asyncio
+async def test_status_hidden_metric_is_clamped_to_zero(mock_app):
+ status_handler = next(
+ r.handler
+ for r in mock_app.get_routes()
+ if r.path == "/api/v1/lxmf/propagation-node/status"
+ )
+ sync_handler = next(
+ r.handler
+ for r in mock_app.get_routes()
+ if r.path == "/api/v1/lxmf/propagation-node/sync"
+ )
+
+ mock_app.current_context.message_router.get_outbound_propagation_node.return_value = (
+ b"somehash"
+ )
+ mock_app.current_context.message_router.propagation_transfer_last_result = 1
+
+ with (
+ patch.object(
+ mock_app.current_context.database.messages,
+ "count_lxmf_messages",
+ side_effect=[10, 20],
+ ),
+ patch.object(
+ mock_app.current_context.database.messages,
+ "count_lxmf_messages_by_state",
+ side_effect=[2, 8],
+ ),
+ ):
+ await sync_handler(None)
+ response = await status_handler(None)
+
+ data = json.loads(response.body)["propagation_node_status"]
+ assert data["messages_hidden"] == 0
+
+
+def test_on_lxmf_sending_failed_forwards_context_to_state_update(mock_app):
+ msg = MagicMock(spec=LXMF.LXMessage)
+ msg.state = LXMF.LXMessage.FAILED
+ msg.try_propagation_on_fail = False
+
+ with patch.object(mock_app, "on_lxmf_sending_state_updated") as state_update_mock:
+ mock_app.on_lxmf_sending_failed(msg, context=mock_app.current_context)
+
+ state_update_mock.assert_called_once_with(msg, context=mock_app.current_context)
diff --git a/tests/backend/test_markdown_renderer.py b/tests/backend/test_markdown_renderer.py
index 4f69aaee..ddacae05 100644
--- a/tests/backend/test_markdown_renderer.py
+++ b/tests/backend/test_markdown_renderer.py
@@ -1,4 +1,5 @@
import unittest
+import time
from meshchatx.src.backend.markdown_renderer import MarkdownRenderer
@@ -22,7 +23,6 @@ class TestMarkdownRenderer(unittest.TestCase):
self.assertIn("<pre", rendered)
self.assertIn("<code", rendered)
self.assertIn("language-python", rendered)
- # Check for escaped characters
self.assertTrue(
"print('hello')" in rendered
or "print('hello')" in rendered,
@@ -67,6 +67,123 @@ class TestMarkdownRenderer(unittest.TestCase):
self.assertIn("Para 1", rendered)
self.assertIn("Para 2", rendered)
+ def test_render_none_and_empty(self):
+ self.assertEqual(MarkdownRenderer.render(None), "")
+ self.assertEqual(MarkdownRenderer.render(""), "")
+ r = MarkdownRenderer.render(" ")
+ self.assertIsInstance(r, str)
+
+ def test_xss_script_tags(self):
+ cases = [
+ "<script>alert(1)</script>",
+ "<SCRIPT>alert(1)</SCRIPT>",
+ "<script src=\"evil.js\"></script>",
+ "</script><script>alert(1)</script>",
+ ]
+ for s in cases:
+ r = MarkdownRenderer.render(s)
+ self.assertNotIn("<script>", r, msg=s)
+ self.assertNotIn("</script>", r, msg=s)
+ self.assertIn("<", r, msg=s)
+
+ def test_xss_event_handlers(self):
+ cases = [
+ '<img src="x" onerror="alert(1)">',
+ '<body onload="alert(1)">',
+ '<a href="x" onmouseover="alert(1)">x</a>',
+ '<svg onload="alert(1)">',
+ ]
+ for s in cases:
+ r = MarkdownRenderer.render(s)
+ self.assertNotIn("<img", r, msg=s)
+ self.assertNotIn("<body", r, msg=s)
+ self.assertNotIn("<svg", r, msg=s)
+ self.assertNotIn("<a ", r, msg=s)
+ self.assertIn("<", r, msg=s)
+
+ def test_xss_link_href_javascript(self):
+ r = MarkdownRenderer.render("[click](javascript:alert(1))")
+ self.assertNotIn("javascript:", r)
+ self.assertIn('href="#"', r)
+
+ def test_xss_link_href_data(self):
+ r = MarkdownRenderer.render("[click](data:text/html,<script>alert(1)</script>)")
+ self.assertNotIn("data:", r)
+ self.assertIn('href="#"', r)
+
+ def test_xss_link_href_vbscript(self):
+ r = MarkdownRenderer.render("[click](vbscript:msgbox(1))")
+ self.assertNotIn("vbscript:", r)
+ self.assertIn('href="#"', r)
+
+ def test_safe_links_preserved(self):
+ r = MarkdownRenderer.render("[link](https://example.com/path)")
+ self.assertIn('href="https://example.com/path"', r)
+ r = MarkdownRenderer.render("[link](/relative)")
+ self.assertIn('href="/relative"', r)
+ r = MarkdownRenderer.render("[link](#anchor)")
+ self.assertIn('href="#anchor"', r)
+
+ def test_redos_safe_repeated_markers(self):
+ t0 = time.perf_counter()
+ MarkdownRenderer.render("*" * 8000)
+ MarkdownRenderer.render("#" * 8000)
+ MarkdownRenderer.render("`" * 8000)
+ MarkdownRenderer.render("[](" * 2000 + "x" * 2000 + ")" * 2000)
+ elapsed = time.perf_counter() - t0
+ self.assertLess(elapsed, 2.0, "ReDoS or excessive backtracking suspected")
+
+ def test_malformed_unclosed_markdown(self):
+ cases = [
+ "**bold no close",
+ "```\ncode no close",
+ "*italic",
+ "`code",
+ "___under",
+ "[unclosed link](url",
+ "![unclosed img](src",
+ ]
+ for s in cases:
+ r = MarkdownRenderer.render(s)
+ self.assertIsInstance(r, str)
+ self.assertNotIn("<script>", r)
+
+ def test_very_long_input(self):
+ big = "x" * (1024 * 1024)
+ r = MarkdownRenderer.render(big)
+ self.assertIsInstance(r, str)
+ self.assertIn("x", r)
+
+ def test_unicode_and_control_chars(self):
+ cases = [
+ "\x00hello",
+ "hello\x07world",
+ "\u202eRTL",
+ "\ufffd replacement",
+ "# header \n normal",
+ ]
+ for s in cases:
+ r = MarkdownRenderer.render(s)
+ self.assertIsInstance(r, str)
+
+ def test_render_returns_string(self):
+ r = MarkdownRenderer.render("Hello **world**")
+ self.assertIsInstance(r, str)
+ self.assertIn("Hello", r)
+ self.assertIn("strong", r)
-if __name__ == "__main__":
- unittest.main()
+ def test_message_content_style_input_safe(self):
+ """Simulate message body (e.g. decoded from LXMF) passed to render; must not crash or emit script."""
+ cases = [
+ b"Hello world".decode("utf-8"),
+ ("Hi \ufffd replacement char " * 10).strip(),
+ "\x00\x01\n\t\r",
+ "<script>alert(1)</script> normal text",
+ "**bold** and [link](javascript:x) end",
+ "\u202eRTL override",
+ ]
+ for s in cases:
+ r = MarkdownRenderer.render(s)
+ self.assertIsInstance(r, str)
+ self.assertNotIn("<script>", r)
+ self.assertNotIn("javascript:", r)
diff --git a/tests/backend/test_meshchat_coverage.py b/tests/backend/test_meshchat_coverage.py
index 28d3aafb..fcb5c121 100644
--- a/tests/backend/test_meshchat_coverage.py
+++ b/tests/backend/test_meshchat_coverage.py
@@ -290,4 +290,4 @@ def test_on_lxmf_sending_failed_no_propagation(mock_app):
mock_app.on_lxmf_sending_state_updated = MagicMock()
mock_app.on_lxmf_sending_failed(mock_msg)
- mock_app.on_lxmf_sending_state_updated.assert_called_once_with(mock_msg)
+ mock_app.on_lxmf_sending_state_updated.assert_called_once_with(mock_msg, context=None)
diff --git a/tests/backend/test_notifications.py b/tests/backend/test_notifications.py
index ca57f047..d1df2c51 100644
--- a/tests/backend/test_notifications.py
+++ b/tests/backend/test_notifications.py
@@ -264,9 +264,6 @@ async def test_notifications_api(mock_app):
# But it's defined as a nested function.
# Alternatively, we can test the DAOs and meshchat.py logic that the handler uses.
- # For now, let's verify that system notifications are correctly combined with LXMF messages.
- # This is done in notifications_get.
-
# Let's test a spike of notifications
for i in range(100):
mock_app.database.misc.add_notification(
diff --git a/tests/backend/test_performance_bottlenecks.py b/tests/backend/test_performance_bottlenecks.py
index 89f10814..909a1a87 100644
--- a/tests/backend/test_performance_bottlenecks.py
+++ b/tests/backend/test_performance_bottlenecks.py
@@ -90,7 +90,6 @@ class TestPerformanceBottlenecks(unittest.TestCase):
app_data = b"app_data"
packet_hash = b"packet_hash"
- # This is the synchronous part that could be a bottleneck
self.announce_manager.upsert_announce(
self.reticulum_mock,
identities[i],
diff --git a/tests/backend/test_property_based.py b/tests/backend/test_property_based.py
index 677d18c4..38827969 100644
--- a/tests/backend/test_property_based.py
+++ b/tests/backend/test_property_based.py
@@ -171,7 +171,6 @@ def test_parse_bool_query_param(val):
@given(data=st.binary())
def test_parse_lxmf_display_name_robustness(data):
- # This should never crash
try:
parse_lxmf_display_name(data)
except Exception as e:
@@ -180,7 +179,6 @@ def test_parse_lxmf_display_name_robustness(data):
@given(data=st.binary())
def test_parse_lxmf_propagation_node_app_data_robustness(data):
- # This should never crash
try:
result = parse_lxmf_propagation_node_app_data(data)
if result is not None:
@@ -226,7 +224,6 @@ def test_interface_config_parser_best_effort_property(names, keys, values):
@given(data=st.binary())
def test_parse_lxmf_stamp_cost_robustness(data):
- # This should never crash
try:
parse_lxmf_stamp_cost(data)
except Exception as e:
@@ -235,7 +232,6 @@ def test_parse_lxmf_stamp_cost_robustness(data):
@given(name=st.text())
def test_parse_nomadnetwork_node_display_name_robustness(name):
- # This should never crash
try:
parse_nomadnetwork_node_display_name(name)
except Exception as e:
@@ -244,7 +240,6 @@ def test_parse_nomadnetwork_node_display_name_robustness(name):
@given(packed=st.binary())
def test_telemeter_from_packed_robustness(packed):
- # This should never crash
try:
Telemeter.from_packed(packed)
except Exception as e:
@@ -444,8 +439,6 @@ def test_markdown_renderer_headers(content):
assert result.startswith("<h1")
assert result.endswith("</h1>")
- # If the content doesn't contain markdown special chars, we can expect it to be there escaped
- # This is a safer assertion for property-based testing
if not any(c in content for c in "*_~`[]()"):
assert html.escape(content) in result
@@ -852,13 +845,11 @@ def test_lxmf_utils_conversions_robustness(
hex_str=st.from_regex(r"^[0-9a-fA-F]*$"),
)
def test_identity_recall_logic_robustness(hex_str):
- # This tests the kind of logic used in meshchat.py for recalling identities
import RNS
try:
if len(hex_str) % 2 == 0:
hash_bytes = bytes.fromhex(hex_str)
- # Just ensure RNS doesn't crash on random bytes
RNS.Identity.recall(hash_bytes)
except Exception:
pass
diff --git a/tests/backend/test_reply_detection.py b/tests/backend/test_reply_detection.py
index 1408e2fb..0f70427f 100644
--- a/tests/backend/test_reply_detection.py
+++ b/tests/backend/test_reply_detection.py
@@ -31,7 +31,6 @@ def test_fuzz_reply_detection_no_crash(content):
mock_msg.snr = 0
mock_msg.q = 0
- # This will trigger the detection logic
mesh_chat.db_upsert_lxmf_message(mock_msg)
diff --git a/tests/backend/test_risky_inputs.py b/tests/backend/test_risky_inputs.py
new file mode 100644
index 00000000..bbff2a47
--- /dev/null
+++ b/tests/backend/test_risky_inputs.py
@@ -0,0 +1,205 @@
+import base64
+import json
+import unittest
+
+
+class TestBase64DecodeRisky(unittest.TestCase):
+ def test_invalid_base64_does_not_crash_display_name(self):
+ try:
+ from meshchatx.src.backend.meshchat_utils import parse_lxmf_display_name
+ except ImportError:
+ self.skipTest("meshchat_utils not importable")
+ for bad in ["!!!", "\x00\x01", "a" * 10000, " ", "=", "===", "a\nb\tc", "\xff\xfe"]:
+ result = parse_lxmf_display_name(bad, default_value="Fallback")
+ self.assertIsInstance(result, str)
+
+ def test_invalid_base64_does_not_crash_propagation_node_app_data(self):
+ try:
+ from meshchatx.src.backend.meshchat_utils import (
+ parse_lxmf_propagation_node_app_data,
+ )
+ except ImportError:
+ self.skipTest("meshchat_utils not importable")
+ for bad in ["!!!", "", "a" * 5000]:
+ try:
+ parse_lxmf_propagation_node_app_data(bad)
+ except Exception:
+ pass
+
+ def test_invalid_base64_does_not_crash_stamp_cost(self):
+ try:
+ from meshchatx.src.backend.meshchat_utils import parse_lxmf_stamp_cost
+ except ImportError:
+ self.skipTest("meshchat_utils not importable")
+ for bad in ["!!!", "", "x" * 10000]:
+ try:
+ parse_lxmf_stamp_cost(bad)
+ except Exception:
+ pass
+
+ def test_base64_decode_direct_invalid_chars(self):
+ for s in ["a\x00b", "a--b", "a/b+c", " " * 1000]:
+ try:
+ base64.b64decode(s, validate=False)
+ except Exception:
+ pass
+ try:
+ base64.b64decode("!!!", validate=True)
+ except Exception:
+ pass
+
+
+class TestJsonLoadsRisky(unittest.TestCase):
+ def test_deeply_nested_json_does_not_crash_message_dict(self):
+ try:
+ from meshchatx.src.backend.lxmf_utils import convert_db_lxmf_message_to_dict
+ except ImportError:
+ self.skipTest("lxmf_utils not importable")
+ deep = "x"
+ for _ in range(200):
+ deep = '{"a":' + deep + "}"
+ db_msg = {
+ "id": 1,
+ "hash": "h",
+ "source_hash": "s",
+ "destination_hash": "d",
+ "is_incoming": True,
+ "state": "delivered",
+ "progress": 100.0,
+ "method": "direct",
+ "delivery_attempts": 0,
+ "next_delivery_attempt_at": None,
+ "title": "t",
+ "content": "c",
+ "fields": deep,
+ "timestamp": 123.0,
+ "rssi": -50,
+ "snr": 10,
+ "quality": 100,
+ "is_spam": False,
+ "created_at": "2024-01-01T00:00:00Z",
+ "updated_at": "2024-01-01T00:00:00Z",
+ }
+ try:
+ convert_db_lxmf_message_to_dict(db_msg)
+ except (RecursionError, OverflowError):
+ pass
+
+ def test_prototype_pollution_keys_in_fields_do_not_crash(self):
+ try:
+ from meshchatx.src.backend.lxmf_utils import convert_db_lxmf_message_to_dict
+ except ImportError:
+ self.skipTest("lxmf_utils not importable")
+ db_msg = {
+ "id": 1,
+ "hash": "h",
+ "source_hash": "s",
+ "destination_hash": "d",
+ "is_incoming": True,
+ "state": "delivered",
+ "progress": 100.0,
+ "method": "direct",
+ "delivery_attempts": 0,
+ "next_delivery_attempt_at": None,
+ "title": "t",
+ "content": "c",
+ "fields": '{"__proto__":{"x":1},"constructor":{"prototype":{"y":1}}}',
+ "timestamp": 123.0,
+ "rssi": -50,
+ "snr": 10,
+ "quality": 100,
+ "is_spam": False,
+ "created_at": "2024-01-01T00:00:00Z",
+ "updated_at": "2024-01-01T00:00:00Z",
+ }
+ result = convert_db_lxmf_message_to_dict(db_msg)
+ self.assertIsInstance(result, dict)
+
+ def test_huge_fields_string_does_not_hang(self):
+ try:
+ from meshchatx.src.backend.lxmf_utils import convert_db_lxmf_message_to_dict
+ except ImportError:
+ self.skipTest("lxmf_utils not importable")
+ db_msg = {
+ "id": 1,
+ "hash": "h",
+ "source_hash": "s",
+ "destination_hash": "d",
+ "is_incoming": True,
+ "state": "delivered",
+ "progress": 100.0,
+ "method": "direct",
+ "delivery_attempts": 0,
+ "next_delivery_attempt_at": None,
+ "title": "t",
+ "content": "c",
+ "fields": '{"k":"' + "v" * 50000 + '"}',
+ "timestamp": 123.0,
+ "rssi": -50,
+ "snr": 10,
+ "quality": 100,
+ "is_spam": False,
+ "created_at": "2024-01-01T00:00:00Z",
+ "updated_at": "2024-01-01T00:00:00Z",
+ }
+ try:
+ convert_db_lxmf_message_to_dict(db_msg)
+ except (MemoryError, json.JSONDecodeError):
+ pass
+
+ def test_invalid_json_fields_returns_safe_dict(self):
+ try:
+ from meshchatx.src.backend.lxmf_utils import convert_db_lxmf_message_to_dict
+ except ImportError:
+ self.skipTest("lxmf_utils not importable")
+ db_msg = {
+ "id": 1,
+ "hash": "h",
+ "source_hash": "s",
+ "destination_hash": "d",
+ "is_incoming": True,
+ "state": "delivered",
+ "progress": 100.0,
+ "method": "direct",
+ "delivery_attempts": 0,
+ "next_delivery_attempt_at": None,
+ "title": "t",
+ "content": "c",
+ "fields": "not json at all {{{",
+ "timestamp": 123.0,
+ "rssi": -50,
+ "snr": 10,
+ "quality": 100,
+ "is_spam": False,
+ "created_at": "2024-01-01T00:00:00Z",
+ "updated_at": "2024-01-01T00:00:00Z",
+ }
+ result = convert_db_lxmf_message_to_dict(db_msg)
+ self.assertIsInstance(result, dict)
+ self.assertIn("fields", result)
+
+
+class TestPathAndStringRisky(unittest.TestCase):
+ def test_interface_config_with_null_byte(self):
+ try:
+ from meshchatx.src.backend.interface_config_parser import (
+ InterfaceConfigParser,
+ )
+ except ImportError:
+ self.skipTest("interface_config_parser not importable")
+ try:
+ InterfaceConfigParser.parse("[section]\nname\x00=value")
+ except Exception:
+ pass
+
+ def test_interface_config_very_long_line(self):
+ try:
+ from meshchatx.src.backend.interface_config_parser import (
+ InterfaceConfigParser,
+ )
+ except ImportError:
+ self.skipTest("interface_config_parser not importable")
+ try:
+ InterfaceConfigParser.parse("[x]\nkey=" + "v" * 100000)
+ except Exception:
+ pass
diff --git a/tests/backend/test_security_fuzzing.py b/tests/backend/test_security_fuzzing.py
index b3d46308..6dedd909 100644
--- a/tests/backend/test_security_fuzzing.py
+++ b/tests/backend/test_security_fuzzing.py
@@ -355,7 +355,6 @@ def test_telephone_callback_fuzzing(mock_app, caller_id_bytes):
)
def test_message_dao_upsert_fuzzing(mock_app, data):
"""Fuzz MessageDAO.upsert_lxmf_message with varied dictionary data."""
- # This should not raise SQL errors or crash
mock_app.database.messages.upsert_lxmf_message(data)
diff --git a/tests/backend/test_telemetry_extended.py b/tests/backend/test_telemetry_extended.py
index 5550149c..9342a316 100644
--- a/tests/backend/test_telemetry_extended.py
+++ b/tests/backend/test_telemetry_extended.py
@@ -40,9 +40,4 @@ def test_telemeter_unpack_location_robustness():
def test_sideband_request_format_compatibility():
- # Sideband telemetry request command is 0x01
- # It can be a simple int 0x01 or a dict {0x01: timebase}
-
- # This test is more about the logic in on_lxmf_delivery,
- # but we can verify our assumptions about command structure here if needed.
pass
diff --git a/tests/backend/test_telemetry_integration.py b/tests/backend/test_telemetry_integration.py
index 12e69118..8dbbefd4 100644
--- a/tests/backend/test_telemetry_integration.py
+++ b/tests/backend/test_telemetry_integration.py
@@ -65,7 +65,6 @@ async def test_process_incoming_telemetry_single(mock_app):
@pytest.mark.asyncio
async def test_process_incoming_telemetry_stream(mock_app):
- # This simulates receiving a telemetry stream (e.g. from Sideband collector)
entries = [
(
"peer1",
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────